//
// Created by 郝羽 on 2018/6/7.
//

#include "batch_sender.hpp"

#include <glib-2.0/glib.h>
#include <list>
#include <climits>

#include "xm_log.h"
#include "xcutil.h"

static constexpr int kBatchMaxMsgs = IOV_MAX;

BatchSender* BatchSenderNew(const char* id, int packets, int duration) {
    auto sender = new BatchSender();
    sender->id = g_strdup(id);
    sender->packets = packets;
    sender->duration = duration;
    return sender;
}

void BatchSenderDelete(BatchSender* sender) {
    dbgi("deleting batch sender |%s|, %lu messages remains", sender->id,
         sender->messages.size());
    for (auto msg : sender->messages) {
        if (msg.buffer != nullptr) {
            g_slice_free1(msg.buffer_size, msg.buffer);
        }
    }

    sender->messages.clear();

    if (sender->id) {
        g_free(sender->id);
    }

    delete sender;
}

void BatchSenderAppend(BatchSender* sender, const unsigned char* buf, int buf_size) {
    BatchMessage message;
    message.buffer = (unsigned char*)g_slice_alloc(buf_size);
    message.buffer_size = buf_size;
    message.start_time = get_timestamp_ms();

    memcpy(message.buffer, buf, buf_size);

    sender->messages.emplace_back(message);
}

static bool shouldSend(BatchSender* sender) {
    if (sender->messages.empty()) {
        return false;
    }
    int64_t now = get_timestamp_ms();

    auto head = *sender->messages.begin();

    return sender->messages.size() >= sender->packets
           || (now - head.start_time) > sender->duration;
}

static void takeMessages(BatchSender* sender, std::vector<BatchMessage>& messages) {
    while (sender->messages.size() > 0 && messages.size() < kBatchMaxMsgs) {
        auto message = sender->messages.front();
        sender->messages.pop_front();
        messages.emplace_back(message);
    }
}

bool BatchSenderGetMessages(BatchSender* sender, std::vector<BatchMessage>& messages) {
    if (!shouldSend(sender)) {
        return false;
    }

    takeMessages(sender, messages);

    return shouldSend(sender);
}

void BatchSenderFreeMessages(BatchSender* sender, std::vector<BatchMessage>& messages) {
    takeMessages(sender, messages);

    for (size_t i = 0; i < messages.size(); ++i) {
        BatchMessage& message = messages[i];
        g_slice_free1(message.buffer_size, message.buffer);
    }

    messages.clear();

    return;
}